package com.zlm.realtime.utils

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import java.util.Properties

/**
 * @author Harbour 
 * @date 2021-04-09 15:03
 */
object MyKafkaSinkUtils {
    var kafkaProducer: KafkaProducer[String, String] = null

    def createKafkaProducer: KafkaProducer[String, String] = {
        val properties = new Properties
        properties.put("bootstrap.servers",
            MyPropertiesUtils.getProperty("kafka.broker.list"))
        properties.put("key.serializer",
            MyPropertiesUtils.getProperty("key.serializer"))
        properties.put("value.serializer",
            MyPropertiesUtils.getProperty("value.serializer"))
        properties.put("enable.idempotence",
            MyPropertiesUtils.getProperty("enable.idempotence"))

        var producer: KafkaProducer[String, String] = null
        try
            producer = new KafkaProducer[String, String](properties)
        catch {
            case e: Exception => e.printStackTrace()
        }
        producer
    }

    def send(topic: String, msg: String): Unit = {
        if (kafkaProducer == null)
            kafkaProducer = createKafkaProducer
        kafkaProducer.send(new ProducerRecord[String, String](topic, msg))
    }

    def send(topic: String, key:String, msg: String): Unit = {
        if (kafkaProducer == null) kafkaProducer = createKafkaProducer
        kafkaProducer.send(new ProducerRecord[String, String](topic, key, msg))
    }

}
